package com.gitee.sparta.core.process.operate.redis;

import com.gitee.sparta.common.SpartaConstant;
import com.gitee.sparta.common.SpartaLuaConstant;
import com.gitee.sparta.common.helpers.*;
import com.gitee.sparta.core.bean.metadata.Metadata;
import com.gitee.sparta.core.process.operate.MetadataTabOperate;
import com.gitee.sparta.core.property.SpartaCoreProperties;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RScript;
import org.redisson.api.RedissonClient;
import org.redisson.client.codec.StringCodec;

import java.util.*;

/**
 * 使用 lua 脚本针对 metadata_tab 中进行操作
 *
 * @param <V> 消息体泛型
 * @author zyred
 * @since v 0.0.1
 */
@Slf4j
public class LuaMetadataTabOperate<V> implements MetadataTabOperate<V> {

    private final RedissonClient redissonClient;
    private final SpartaCoreProperties properties;

    public LuaMetadataTabOperate(RedissonClient redissonClient,
                                 SpartaCoreProperties properties) {
        this.redissonClient = redissonClient;
        this.properties = properties;
    }

    @Override
    @SuppressWarnings("all")
    public Metadata<V> readMetadata(String fullTopic) {
        Map<String, String> cache = SpartaLuaConstant.getSha(SpartaConstant.READ_METADATA);

        final String metadataTabKey = RedisKeyHelper.tabKey(
                this.properties.getSystemName(),
                SpartaConstant.TABLE_METADATA_NAME
        );

        final List<Object> keys = Collections.singletonList(metadataTabKey);

        // topic:metaId-0
        if (fullTopic.contains(SpartaConstant.LINE)) {
            String[] split = fullTopic.split(SpartaConstant.LINE);
            fullTopic = split[0];
        }

        try {
            RScript script = this.redissonClient.getScript(new StringCodec());
            Object result = script.evalSha(RScript.Mode.READ_ONLY,
                    cache.get(SpartaConstant.LUA_SHA_KEY), RScript.ReturnType.VALUE, keys, fullTopic);

            if (log.isDebugEnabled()) {
                log.debug("[Sparta] Read metadata：{}, topic：{}", result, fullTopic);
            }
            if (Objects.nonNull(result)) {
                return (Metadata<V>) JacksonHelper.deserialize((String) result, Metadata.class);
            }
            return null;
        } catch (Exception ex) {
            log.error("[Sparta] Read metadata failed to execute, topic: {}, exception：{}",
                    fullTopic, ex.getMessage(), ex);
            ScriptShaHelper.reloadScriptSha(ex, redissonClient, properties);
            return null;
        }
    }

    @Override
    public void delete(final String readableTabKey,
                       final String timeoutTabKey,
                       final String metadataTabKey,
                       final String segmentFields,
                       final String field) {
        Map<String, String> cache = SpartaLuaConstant.getSha(SpartaConstant.DELETE);
        List<Object> keys = Arrays.asList(readableTabKey, timeoutTabKey, metadataTabKey);

        try {
            RScript script = this.redissonClient.getScript(new StringCodec());
            script.evalSha(RScript.Mode.READ_ONLY, cache.get(SpartaConstant.LUA_SHA_KEY),
                    RScript.ReturnType.VALUE, keys, field, segmentFields);
        } catch (Exception ex) {
            log.error("[Sparta] Delete failed to execute, topic: {}, exception：{}",
                    field, ex.getMessage(), ex);
            ScriptShaHelper.reloadScriptSha(ex, redissonClient, properties);
        }
    }

    @Override
    @SuppressWarnings("all")
    public Map<String, Metadata<V>> readMetadataMap(String listenTopic) {
        Map<String, String> cache = SpartaLuaConstant.getSha(SpartaConstant.READ_METADATA_MAP);
        final String metadataTabKey = RedisKeyHelper.tabKey(
                this.properties.getSystemName(),
                SpartaConstant.TABLE_METADATA_NAME
        );

        final String readableTabKey = RedisKeyHelper.readableKey(
                this.properties.getSystemName(),
                this.properties.getIsolation(),
                listenTopic
        );
        List<Object> keys = Arrays.asList(readableTabKey, metadataTabKey);

        try {
            RScript script = this.redissonClient.getScript(new StringCodec());
            Object result = script.evalSha(RScript.Mode.READ_ONLY, cache.get(SpartaConstant.LUA_SHA_KEY),
                    RScript.ReturnType.MAPVALUELIST, keys, SpartaConstant.LINE);
            if (result instanceof List) {
                List<Object> resultList = (List<Object>) result;
                String topics = String.valueOf(resultList.get(0));
                List<String> topicList = CollHelper.convertList(topics, SpartaConstant.COMMA);

                List<String> metadataList = (List<String>)resultList.get(1);

                Map<String, Metadata<V>> mapping = new HashMap<>();
                for (String compressMetadata : metadataList) {
                    if (Objects.isNull(compressMetadata)) {
                        continue;
                    }
                    if (this.properties.isEnableMetadataCompress()) {
                        compressMetadata = ZipStrHelper.unCompress(compressMetadata);
                    }
                    Metadata<V> metadata = (Metadata<V>) JacksonHelper.deserialize(compressMetadata, Metadata.class);
                    mapping.put(RedisKeyHelper.fieldName(metadata.getTopic(), metadata.getMetaId()), metadata);
                }


                Map<String, Metadata<V>> metadataMap = new HashMap<>(topicList.size());
                for (String topic : topicList) {
                    if (topic.contains(SpartaConstant.LINE)) {
                        int index = topic.indexOf(SpartaConstant.LINE);
                        String sub = topic.substring(0, index);
                        metadataMap.put(topic, mapping.get(sub));
                    } else {
                        metadataMap.put(topic, mapping.get(topic));
                    }
                }
                return metadataMap;
            }
            return Collections.emptyMap();
        } catch ( Exception ex) {
            log.error("[Sparta] Read metadata map, failed to execute, exception：{}", ex.getMessage(), ex);
            ScriptShaHelper.reloadScriptSha(ex, redissonClient, properties);
            return Collections.emptyMap();
        }
    }

    @SuppressWarnings("all")
    @Override
    public Metadata<V> readDelMetadata(String fieldName) {
        Map<String, String> cache = SpartaLuaConstant.getSha(SpartaConstant.READ_DEL_METADATA_MAP);

        final String metadataTabKey = RedisKeyHelper.tabKey(
                this.properties.getSystemName(),
                SpartaConstant.TABLE_METADATA_NAME
        );

        final List<Object> keys = Collections.singletonList(metadataTabKey);

        try {
            RScript script = this.redissonClient.getScript(new StringCodec());
            Object result = script.evalSha(RScript.Mode.READ_ONLY,
                    cache.get(SpartaConstant.LUA_SHA_KEY), RScript.ReturnType.VALUE, keys, fieldName);

            if (log.isDebugEnabled()) {
                log.debug("[Sparta] Read del metadata：{}, topic：{}", result, fieldName);
            }
            if (Objects.nonNull(result)) {
                return (Metadata<V>) JacksonHelper.deserialize((String) result, Metadata.class);
            }
            return null;
        } catch (Exception ex) {
            log.error("[Sparta] Read del metadata failed to execute, topic: {}, exception：{}",
                    fieldName, ex.getMessage(), ex);
            ScriptShaHelper.reloadScriptSha(ex, redissonClient, properties);
            return null;
        }
    }
}
